package com.atguigu.tingshu.order.api;

import com.atguigu.tingshu.common.result.Result;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.annotations.Param;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * @author: atguigu
 * @create: 2024-04-24 10:29
 */
@Slf4j
@RestController
@RequestMapping("/api/delay")
public class DelayMessageController {


    @Autowired
    private RedissonClient redissonClient;

    /**
     * 仅用于测试，发送延迟消息
     *
     * @param msg 消息
     * @param ttl 延迟时间单位：秒
     * @return
     */
    @GetMapping("/sendDelayMsg/{msg}/{ttl}")
    public Result sendDelayMsg(@PathVariable String msg, @PathVariable int ttl) {
        //1.基于Redisson客户端对象创建阻塞队列
        RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue("test-queue");
        //2.基于Redisson客户端对象将阻塞队列作为参数创建延迟队列
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        //3.调用延迟队列对象发送延迟消息
        delayedQueue.offer(msg, ttl, TimeUnit.SECONDS);
        log.info("发送延迟消息成功：{}，延迟时间：{}s", msg, ttl);
        return Result.build(null, 200, "延迟消息发送成功");
    }


    /**
     * 消费者：监听延迟消息
     */
    @PostConstruct
    public void receiveDelayMsg() {
        //1.基于Redisson客户端对象创建阻塞队列
        RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue("test-queue");
        new Thread(() -> {
            while (true) {
                String poll = null;
                try {
                    poll = blockingQueue.poll(5, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                if (StringUtils.isNotBlank(poll)) {
                    log.info("监听延迟消息:{}", poll);
                }
            }
        }).start();
    }

}
